package com.bigdata.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.*;
import java.net.URI;
import java.util.Arrays;

public class HDFSDemo {
    private Configuration conf;
    private FileSystem fs;


    @Before
    public void initFileSystem() throws Exception {
        /**
         * 在写这些调用api的代码时，可以参考源码，记住几行关键的代码
         * 写的时候根据源码的提示，连蒙带猜
         */
        conf = new Configuration(true);
        // fs 就是我们连接HDFS的对象
        // fs = FileSystem.get(URI.create("hdfs://node01:8020"), conf);
        fs = FileSystem.get(URI.create("hdfs://node01:9000"), conf);
    }

    @Test
    public void mkdirTest() throws IOException {
        System.out.println(fs);

        // 参数填写目录路径，如果目录存在，先删除再创建

        Path dirPath = new Path("/bigdata");

        if(fs.exists(dirPath)) {
            fs.delete(dirPath, true);
        }

        boolean flag = fs.mkdirs(dirPath);
        System.out.println(flag);
    }

    @Test
    public void createTest() throws IOException {
        // System.out.println(fs);

        // 参数填写目录路径
        Path filePath = new Path("/bigdata/1.txt");

        // 创建，建立IO连接
        FSDataOutputStream outputStream = fs.create(filePath);

        String str = "TES新阵容德玛西亚杯开门红，轻取UP！网友：小奶油未来可期？";
        // 写入内容到指定文件中
        outputStream.write(str.getBytes());

        outputStream.close();
    }

    /**
     * 删除文件
     *
     * @throws IOException
     */
    @Test
    public void deleteTest() throws IOException {
        // Path delPath = new Path("/bigdata/1.txt");
        Path delPath = new Path("/bigdata");

        // 如果是空目录， 可以直接删除， 不是空目录，第二个参数要设置为true
        System.out.println(fs.delete(delPath, true));
    }


    /**
     * 上传文件
     *
     * @throws IOException
     */
    @Test
    public void uploadTest() throws IOException {
        // 参数填写目录路径
        Path filePath = new Path("/bigdata/story.txt");

        // 创建，建立IO连接
        FSDataOutputStream outputStream = fs.create(filePath);

        // 使用相对路径，就要知道当前的路径
        // System.out.println(new File(".").getAbsolutePath());

        // IO流的代码
        BufferedInputStream inputStream = new BufferedInputStream(new FileInputStream("data/story.txt"));

        IOUtils.copyBytes(inputStream, outputStream, conf, true);
//        String line;
//        while ((line = br.readLine()) != null) {
//            // System.out.println(line);
//            outputStream.write(line.getBytes("utf-8"));
//        }
//        outputStream.close();
//        br.close();
    }


    /**
     * 上传文件
     *
     * @throws IOException
     */
    @Test
    public void readTest() throws IOException {
        Path path = new Path("/data/wc/input/data.txt");
        // 获取元数据
        FileStatus fileStatus = fs.getFileStatus(path);
        // block 大小
        long blockSize = fileStatus.getBlockSize();
        // 文件大小
        long len = fileStatus.getLen();
        // 文件所属
        String owner = fileStatus.getOwner();
        System.out.println(blockSize / 1024 / 1024);
        System.out.println(len);
        System.out.println(owner);

        // 对一个完整的文件做 wordcount
        // 获取所有block的地址信息
        // 一个 BlockLocation 对象表示的是一个block的存储的位置信息
        BlockLocation[] locations = fs.getFileBlockLocations(path, 0, len);

        // 0,1534,node03,node02, 表示我们针对这个文件所计算，
        // 那么就可以在node03 or node02上来执行计算程序
        for (BlockLocation location : locations) {
            System.out.println("aaa---" + location);
            System.out.println(Arrays.toString(location.getHosts()));
        }

        // 指定位置读取数据, 从头读取
        FSDataInputStream inputStream = fs.open(path);

        /**
         * 在做计算时，可能会读取一个block块的部分数据
         * 所以需要这么一个功能: 设置偏移量读取数据
         */
        inputStream.seek(12);

        byte[] bys = new byte[12];
        inputStream.read(bys);

        System.out.println(new String(bys));
    }


    @After
    public void close() throws IOException {
        fs.close();
    }


}
